package com.flink.datastreamapi.aggregation;

import com.flink.datastreamapi.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class AggregationDemo {

    public static void main(String[] args) throws Exception {
        //获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //数据源
        DataStreamSource<WaterSensor> dataStreamSource = env.fromElements(
                new WaterSensor("sensor_1", 11l, 11),
                new WaterSensor("sensor_1", 12l, 12),
                new WaterSensor("sensor_2", 2l, 2),
                new WaterSensor("sensor_3", 3l, 3));
        //数据转换处理--分组（分区）
        //方式一： 使用匿名类来实现KeySelector
//        KeyedStream<WaterSensor, String> keyedStream1 = dataStreamSource.keyBy(new KeySelector<WaterSensor, String>() {
//            @Override
//            public String getKey(WaterSensor value) throws Exception {
//                return value.id;
//            }
//        });

        //方式二 使用Lambda表达式
        KeyedStream<WaterSensor, String> keyedStream2 = dataStreamSource.keyBy(value -> value.id);

        //聚合--- 最大值 ,根据WaterSensor对象的属性vc来求组内最大值

        //max 与maxby的区别：
        // 		max：只会取比较字段的最大值，非比较字段保留第一次的值
        //          maxby:取比较字段的最大值，同时非比较字段 取 最大值这条数据的值
        //ingleOutputStreamOperator<WaterSensor> maxOfVC = keyedStream2.max("vc");
        SingleOutputStreamOperator<WaterSensor> maxOfVC = keyedStream2.maxBy("vc");

        maxOfVC.print();


        //执行
        env.execute();
    }

}
